package com.stock_t.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.stock_t.entity.FundFlowHistory;
import com.stock_t.mapper.FundFlowHistoryMapper;
import com.stock_t.service.FundFlowHistoryService;
import com.stock_t.service.StockTradingService;
import com.stock_t.utils.EastMoneyFundFlowUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDate;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * 资金流向历史数据Service实现类
 * 实现资金流向数据的业务逻辑处理
 */
@Slf4j
@Service
public class FundFlowHistoryServiceImpl extends ServiceImpl<FundFlowHistoryMapper, FundFlowHistory> 
        implements FundFlowHistoryService {

    @Autowired
    private FundFlowHistoryMapper fundFlowHistoryMapper;

    @Autowired
    private StockTradingService stockTradingService;

    @Override
    public List<FundFlowHistory> getFundFlowByStockCode(String stockCode) {
        log.info("查询股票 {} 的资金流向历史数据", stockCode);
        return fundFlowHistoryMapper.selectByStockCode(stockCode);
    }

    @Override
    public List<FundFlowHistory> getFundFlowByDateRange(String stockCode, LocalDate startDate, LocalDate endDate) {
        log.info("查询股票 {} 在 {} 到 {} 期间的资金流向数据", stockCode, startDate, endDate);
        return fundFlowHistoryMapper.selectByDateRange(stockCode, startDate, endDate);
    }

    @Override
    @Transactional
    public boolean syncFundFlowData() {
        try {
//            List<StockBasic> stocks = stockBasicMapper.selectList(null);
            List<Map<String, String>> stocksListMap = stockTradingService.getStockList();
            List<String> stocks = new ArrayList<>();
            for (Map<String, String> stringStringMap : stocksListMap) {
                stocks.add(stringStringMap.get("stockCode"));
            }
            int total = stocks.size();

            // 创建线程池
            int processors = Runtime.getRuntime().availableProcessors();
            int threadCount = Math.min(processors * 2, 10); // 线程数为CPU核心数的2倍，但不超过10
            ExecutorService executorService = Executors.newFixedThreadPool(threadCount);

            // 创建计数器用于跟踪进度
            AtomicInteger counter = new AtomicInteger(0);

            // 提交任务到线程池
            for (String stockCode : stocks) {
                executorService.submit(() -> {
                    try {
                        // 获取并保存数据
                        fetchAndSaveHistoryData(stockCode);

                        // 更新计数器并打印进度
                        int current = counter.incrementAndGet();
                        synchronized (System.out) {
                            System.out.printf("已处理: %d/%d, 剩余: %d, 进度: %.2f%%\n",
                                    current, total, (total - current), (current * 100.0 / total));
                        }

                        // 每次请求后短暂休眠，避免请求过于频繁
                        Thread.sleep(300);
                    } catch (Exception e) {
                        System.err.println("处理股票 " + stockCode + " 时出错: " + e.getMessage());
                    }
                });
            }

            // 关闭线程池并等待所有任务完成
            executorService.shutdown();
            try {
                // 等待所有任务完成，最多等待1小时
                if (!executorService.awaitTermination(1, TimeUnit.HOURS)) {
                    System.out.println("任务执行超时，强制关闭线程池");
                    executorService.shutdownNow();
                }
            } catch (InterruptedException e) {
                System.err.println("等待任务完成时被中断: " + e.getMessage());
                executorService.shutdownNow();
                Thread.currentThread().interrupt();
            }
            log.info("全量股票的资金流向数据同步完成");
            return true;
        } catch (Exception e) {
            log.error("同步全量股票的资金流向数据失败,{}", e);
            return false;
        }
    }

    @Override
    @Transactional
    public boolean batchSaveFundFlow(List<FundFlowHistory> fundFlowList) {
        if (fundFlowList == null || fundFlowList.isEmpty()) {
            log.warn("批量保存的资金流向数据为空");
            return false;
        }
        
        try {
            log.info("开始批量保存 {} 条资金流向数据", fundFlowList.size());
            boolean result = this.saveBatch(fundFlowList);
            log.info("批量保存资金流向数据完成，结果: {}", result);
            return result;
        } catch (Exception e) {
            log.error("批量保存资金流向数据失败", e);
            return false;
        }
    }

    @Override
    @Transactional
    public int cleanExpiredData(int daysToKeep) {
        if (daysToKeep <= 0) {
            log.warn("保留天数必须大于0");
            return 0;
        }
        
        LocalDate cutoffDate = LocalDate.now().minusDays(daysToKeep);
        log.info("开始清理 {} 之前的资金流向数据", cutoffDate);
        
        int deletedCount = fundFlowHistoryMapper.deleteBeforeDate(cutoffDate);
        log.info("清理完成，共删除 {} 条记录", deletedCount);
        return deletedCount;
    }


    /**
     * 获取并保存指定股票的历史资金流向数据
     *
     * @param stockCode 股票代码
     * @return 保存的记录数
     */
    @Transactional(rollbackFor = Exception.class)
    public int fetchAndSaveHistoryData(String stockCode) {
        try {
            // 获取历史资金流向数据
            List<Map<String, Object>> historyDataList = EastMoneyFundFlowUtil.fetchFundFlowHistoryData(stockCode);
            if (historyDataList == null || historyDataList.isEmpty()) {
                log.warn("未获取到股票{}的历史资金流向数据", stockCode);
                return 0;
            }

            // 转换为实体对象
            List<FundFlowHistory> historyList = historyDataList.stream()
                    .map(dataMap -> EastMoneyFundFlowUtil.convertToFundFlowHistory(stockCode, dataMap))
                    .collect(Collectors.toList());

            // 获取所有日期
            List<LocalDate> dateList = historyList.stream()
                    .map(FundFlowHistory::getTradeDate)
                    .collect(Collectors.toList());

            // 查询已存在的记录
            LambdaQueryWrapper<FundFlowHistory> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.eq(FundFlowHistory::getStockCode, stockCode)
                    .in(FundFlowHistory::getTradeDate, dateList);
            List<FundFlowHistory> existingRecords = list(queryWrapper);

            // 过滤出不存在的记录
            List<LocalDate> existingDates = existingRecords.stream()
                    .map(FundFlowHistory::getTradeDate)
                    .collect(Collectors.toList());

            List<FundFlowHistory> newRecords = historyList.stream()
                    .filter(history -> !existingDates.contains(history.getTradeDate()))
                    .collect(Collectors.toList());

            // 只保存不存在的记录
            if (!newRecords.isEmpty()) {
                saveBatch(newRecords);
                log.info("成功保存股票{}的{}条新增历史资金流向数据", stockCode, newRecords.size());

                return newRecords.size();
            } else {
                log.info("股票{}的历史资金流向数据已存在，无需保存", stockCode);
                return 0;
            }

        } catch (Exception e) {
            log.error("获取并保存股票{}的历史资金流向数据时发生错误", stockCode, e);
            throw e;
        }
    }

}